package org.egomsl.mw.hive;

import org.apache.hadoop.hive.ql.metadata.DefaultStorageHandler;
import org.apache.hadoop.hive.ql.plan.TableDesc;
import org.apache.hadoop.mapred.InputFormat;
import org.apache.hadoop.mapred.OutputFormat;
import org.apache.hadoop.mapred.SequenceFileOutputFormat;
import org.egomsl.mw.mapred.ABaseInputFormat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Map;

public class ABaseStorageHandler extends DefaultStorageHandler {
    private static Logger log = LoggerFactory.getLogger(ABaseStorageHandler.class);

    public Class<? extends InputFormat> getInputFormatClass() {
        return ABaseInputFormat.class;
    }

    public Class<? extends OutputFormat> getOutputFormatClass() {
        return SequenceFileOutputFormat.class;
    }

    public void configureInputJobProperties(
            TableDesc tableDesc, Map<String, String> jobProperties) {

        String tablePath = jobProperties.get(ABaseInputFormat.ABASE_TABLE_PATH_KEY);
        if(tablePath != null) {
            ABaseInputFormat.setTablePath(this.getConf(), tablePath);
        }else {
            log.warn("No table path specified.");
        }
    }

    public void configureOutputJobProperties(
            TableDesc tableDesc, Map<String, String> jobProperties) {
    }

}
